package beautiful.butterfly.drds.data_exchange.report;

import org.apache.commons.lang3.Validate;


/**
 * 这里主要是业务层面的处理
 */
public final class Messages
{
    public static final String stage = "stage";
    public static final String byte_speed = "byteSpeed";
    public static final String record_speed = "recordSpeed";
    public static final String percentage = "percentage";

    public static final String read_succeed_records = "readSucceedRecords";
    public static final String read_succeed_bytes = "readSucceedBytes";

    public static final String read_failed_records = "readFailedRecords";
    public static final String read_failed_bytes = "readFailedBytes";

    public static final String write_received_records = "writeReceivedRecords";
    public static final String write_received_bytes = "writeReceivedBytes";

    public static final String write_failed_records = "writeFailedRecords";
    public static final String write_failed_bytes = "writeFailedBytes";

    public static final String total_read_records = "totalReadRecords";
    public static final String wait_writer_time = "waitWriterTime";
    public static final String wait_reader_time = "waitReaderTime";
    public static final String transformer_used_time = "totalTransformerUsedTime";
    public static final String transformer_succeed_records = "totalTransformerSuccessRecords";
    public static final String transformer_failed_records = "totalTransformerFailedRecords";
    public static final String transformer_filter_records = "totalTransformerFilterRecords";
    public static final String transformer_name_prefix = "usedTimeByTransformer_";
    public static final String total_read_bytes = "totalReadBytes";
    public static final String total_error_records = "totalErrorRecords";
    public static final String total_error_bytes = "totalErrorBytes";
    public static final String write_succeed_records = "writeSucceedRecords";
    public static final String write_succeed_bytes = "writeSucceedBytes";

    public static Message getReportCommunication(Message now, Message old, int totalStage)
    {
        Validate.isTrue(now != null && old != null,
                "为汇报准备的新旧metric不能为null");

        long totalReadRecords = getTotalReadRecords(now);
        long totalReadBytes = getTotalReadBytes(now);
        now.setLong(total_read_records, totalReadRecords);
        now.setLong(total_read_bytes, totalReadBytes);
        now.setLong(total_error_records, getTotalErrorRecords(now));
        now.setLong(total_error_bytes, getTotalErrorBytes(now));
        now.setLong(write_succeed_records, getWriteSucceedRecords(now));
        now.setLong(write_succeed_bytes, getWriteSucceedBytes(now));

        long timeInterval = now.getTimestamp() - old.getTimestamp();
        long sec = timeInterval <= 1000 ? 1 : timeInterval / 1000;
        long bytesSpeed = (totalReadBytes
                - getTotalReadBytes(old)) / sec;
        long recordsSpeed = (totalReadRecords
                - getTotalReadRecords(old)) / sec;

        now.setLong(byte_speed, bytesSpeed < 0 ? 0 : bytesSpeed);
        now.setLong(record_speed, recordsSpeed < 0 ? 0 : recordsSpeed);
        now.setDouble(percentage, now.getLong(stage) / (double) totalStage);

        if (old.getThrowable() != null)
        {
            now.setThrowable(old.getThrowable());
        }

        return now;
    }

    public static long getTotalReadRecords(final Message message)
    {
        return message.getLong(read_succeed_records) +
                message.getLong(read_failed_records);
    }

    public static long getTotalReadBytes(final Message message)
    {
        return message.getLong(read_succeed_bytes) +
                message.getLong(read_failed_bytes);
    }

    public static long getTotalErrorRecords(final Message message)
    {
        return message.getLong(read_failed_records) +
                message.getLong(write_failed_records);
    }

    public static long getTotalErrorBytes(final Message message)
    {
        return message.getLong(read_failed_bytes) +
                message.getLong(write_failed_bytes);
    }

    public static long getWriteSucceedRecords(final Message message)
    {
        return message.getLong(write_received_records) -
                message.getLong(write_failed_records);
    }

    public static long getWriteSucceedBytes(final Message message)
    {
        return message.getLong(write_received_bytes) -
                message.getLong(write_failed_bytes);
    }


}
